package cn.lzd

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by lzd on 2018/12/21.
  */
object ActionDemo {

  def main(args: Array[String]): Unit = {

    val sc = new SparkContext(new SparkConf().setAppName("rdd-action").setMaster("local"))

    demo6(sc)
    /**
      * saveAsSequenceFile
      * saveAsTextFile
      * toke
      * count
      * collect
      * reduce
      */
  }

  /**
    * saveAsSequenceFile
    * 将数据集的元素，以sequencefile的格式，保存到指定的目录下，本地系统，hdfs或者任何其它hadoop支持的文件系统。RDD的元素必须由key-value对组成，并都实现了Hadoop的Writable接口，或隐式可以转换为Writable（Spark包括了基本类型的转换，例如Int，Double，String等等）
    * TODO
    * @param sc
    */
  def demo6(sc: SparkContext): Unit = {
    val lst = Seq(1 -> "hello", 2 -> "hello", 3 -> "hello", 4 -> "hello", 5 -> "hello", 6 -> "hello", 7 -> "hello", 8 -> "hello", 9 -> "hello", 10 -> "hello")
    val rdd1 = sc.parallelize(lst, 1)
    rdd1.saveAsSequenceFile("/C:\\Users\\lzd\\Desktop\\spark-demo")

  }

  /**
    * saveAsTextFile(路径最后一位是文件夹)
    * 将数据集的元素，以textfile的形式，保存到本地文件系统，hdfs或者任何其它hadoop支持的文件系统。Spark将会调用每个元素的toString方法，并将它转换为文件中的一行文本
    *
    * @param sc
    */
  def demo5(sc: SparkContext): Unit = {
    val lst = Seq(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    val rdd1 = sc.parallelize(lst, 1)
    rdd1.saveAsTextFile("/C:\\Users\\lzd\\Desktop\\spark-demo")

  }


  /**
    * toke(n)
    * first() : 返回数据集的第一个元素（类似于take(1)）
    * 返回一个数组，由数据集的前n个元素组成。注意，这个操作目前并非在多个节点上，并行执行，而是Driver程序所在机器，单机计算所有的元素(Gateway的内存压力会增大，需要谨慎使用）
    *
    * @param sc
    */
  def demo4(sc: SparkContext): Unit = {
    val lst = Seq(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    val rdd1 = sc.parallelize(lst, 1)
    rdd1.take(2).foreach(println(_))
  }

  /**
    * count 返回元素个数
    *
    * @param sc
    */
  def demo3(sc: SparkContext): Unit = {
    val lst = Seq(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    val rdd1 = sc.parallelize(lst, 1)
    println(rdd1.count())
  }

  /**
    * collect()
    * 在Driver的程序中，以数组的形式，返回数据集的所有元素。这通常会在使用filter或者其它操作后，返回一个足够小的数据子集再使用。
    *
    * @param sc
    */
  def demo2(sc: SparkContext): Unit = {
    val lst = Seq(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    val rdd1 = sc.parallelize(lst, 1)
    val rdd2 = rdd1.map(_ * 2)

    val collect: Array[Int] = rdd2.collect()
    for (res <- collect) {
      println(res)
    }
  }

  /**
    * reduce(func) :
    * 通过函数func聚集数据集中的所有元素。Func函数接受2个参数，返回一个值。这个函数必须是关联性的，确保可以被正确的并发执行
    *
    * @param sc
    */
  def demo1(sc: SparkContext): Unit = {
    val lst = Seq(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    val rdd1 = sc.parallelize(lst, 1)
    val i: Int = rdd1.reduce(_ + _)
    println(i)
  }
}
